package com.example.node;

import com.example.node.config.NodeConfig;
import com.example.node.message.Request;
import com.example.node.message.Response;
import com.example.node.message.ResponseAllFuture;
import com.example.node.message.ResponseFuture;
import com.example.udp.message.UdpMessage;
import com.example.udp.server.UdpServer;
import com.example.udp.utils.JavaSerializableUtils;
import com.example.websocket.client.SysWebSocketClient;
import com.example.websocket.server.SysWebSocketServer;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/*
 * @className: NodeImpl
 * @author: chentao
 * @date: 2025/10/25 下午5:15
 * @Version: 1.0
 * @description:
 */
public class NodeImpl implements Node {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Long serverPort;

    private NodeConfig nodeConfig;

    private SysWebSocketServer sysWebSocketServer;

    private UdpServer udpServer;

    private Map<Long, SysWebSocketClient> sysWebSocketClients = new ConcurrentHashMap<>();

    private Map<String, ResponseFuture> responseFutureMap = new ConcurrentHashMap<>();

    private Map<String, ResponseAllFuture> responseAllFutureMap = new ConcurrentHashMap<>();

    public NodeImpl(Long serverPort) {
        this.serverPort = serverPort;
        this.nodeConfig = new NodeConfig(serverPort);
        this.sysWebSocketServer = new SysWebSocketServer(this.serverPort.intValue(), new SysWebSocketServer.WebSocketServerListener() {
            @Override
            public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake) {
                SysWebSocketServer.WebSocketServerListener.super.onOpen(webSocket, clientHandshake);
            }

            @Override
            public void onError(WebSocket webSocket, Exception e) {
                SysWebSocketServer.WebSocketServerListener.super.onError(webSocket, e);
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason, boolean remote) {
                SysWebSocketServer.WebSocketServerListener.super.onClose(webSocket, code, reason, remote);
            }

            @Override
            public void onMessage(WebSocket webSocket, String data) {
                SysWebSocketServer.WebSocketServerListener.super.onMessage(webSocket, data);
            }

            @Override
            public void onMessage(WebSocket webSocket, ByteBuffer byteBuffer) {
                onNodeMessage(byteBuffer);
            }
        });
        this.udpServer = new UdpServer(nodeConfig, new UdpServer.UdpServerMessageListener() {
            @Override
            public void onMessage(UdpMessage udpMessage) {
                if (udpMessage.getSenderId().longValue() < nodeConfig.getServerId().longValue() && !sysWebSocketClients.containsKey(udpMessage.getSenderId())) {
                    NodeConfig targetNodeConfig = udpMessage.getNodeConfig();
                    try {
                        SysWebSocketClient sysWebSocketClient = new SysWebSocketClient(targetNodeConfig.getServerIp(), targetNodeConfig.getServerPort().intValue(), nodeConfig.getServerId(), new SysWebSocketClient.WebSocketClientListener() {
                            @Override
                            public void onOpen(ServerHandshake serverHandshake) {
                                SysWebSocketClient.WebSocketClientListener.super.onOpen(serverHandshake);
                            }

                            @Override
                            public void onClose(int code, String reason, boolean remote) {
                                SysWebSocketClient.WebSocketClientListener.super.onClose(code, reason, remote);
                            }

                            @Override
                            public void onError(Exception ex) {
                                SysWebSocketClient.WebSocketClientListener.super.onError(ex);
                            }

                            @Override
                            public void onMessage(String message) {
                                SysWebSocketClient.WebSocketClientListener.super.onMessage(message);
                            }

                            @Override
                            public void onMessage(ByteBuffer byteBuffer) {
                                onNodeMessage(byteBuffer);
                            }
                        });
                        sysWebSocketClient.start();
                        sysWebSocketClients.put(targetNodeConfig.getServerId(), sysWebSocketClient);
                    } catch (Exception e) {
                        logger.warn("websocket client start error", e);
                    }
                }
            }
        });
    }

    @Override
    public void start() throws Exception {
        this.sysWebSocketServer.start();
        this.udpServer.start();
        logger.warn("node server " + nodeConfig.getServerId() + " start success ...");
    }

    @Override
    public void stop() throws Exception {
        for (SysWebSocketClient sysWebSocketClient : this.sysWebSocketClients.values()) {
            sysWebSocketClient.stop();
        }
        this.sysWebSocketServer.stop();
        this.udpServer.stop();
    }

    @Override
    public void onNodeMessage(ByteBuffer byteBuffer) {
        Object message = JavaSerializableUtils.deSerializable(byteBuffer.array());
        if (message instanceof Request) {
            Request request = (Request) message;
            Response response = handleRequest(request);
            if (response.getSenderId() == 0) {
                response.setSenderId(nodeConfig.getServerId());
            }
            if (this.sysWebSocketClients.containsKey(request.getSenderId())) {
                SysWebSocketClient sysWebSocketClient = this.sysWebSocketClients.get(request.getSenderId());
                if (sysWebSocketClient.isOpen()) {
                    sysWebSocketClient.send(JavaSerializableUtils.serializable(response));
                }
            }
            if (this.sysWebSocketServer.getServerKeys().contains(String.valueOf(request.getSenderId()))) {
                WebSocket webSocket = this.sysWebSocketServer.getWebSocket(String.valueOf(request.getSenderId()));
                if (webSocket.isOpen()) {
                    webSocket.send(JavaSerializableUtils.serializable(response));
                }
            }
        }
        if (message instanceof Response) {
            Response response = (Response) message;
            if (this.responseFutureMap.containsKey(response.getRequest().getRequestId())) {
                this.responseFutureMap.remove(response.getRequest().getRequestId()).set(response);
            }
            if (this.responseAllFutureMap.containsKey(response.getRequest().getRequestId())) {
                int size = this.responseAllFutureMap.get(response.getRequest().getRequestId()).put(response);
                if (size == 0) {
                    this.responseAllFutureMap.remove(response.getRequest().getRequestId());
                }
            }
        }
    }

    @Override
    public Response handleRequest(Request request) {
        return new Response(request, true, "ACK");
    }


    @Override
    public Response send(Request request) throws Exception {
        request.setSenderConfig(nodeConfig);
        if (StringUtils.isEmpty(request.getReceiverId())) {
            throw new Exception("request receiverId is null");
        }
        if (request.getReceiverId() == nodeConfig.getServerId()) {
            throw new Exception("request receiverId is self");
        }
        if (this.sysWebSocketClients.containsKey(request.getReceiverId()) || this.sysWebSocketServer.getServerKeys().contains(String.valueOf(request.getReceiverId()))) {
            if (this.sysWebSocketClients.containsKey(request.getReceiverId())) {
                this.sysWebSocketClients.get(request.getReceiverId()).send(JavaSerializableUtils.serializable(request));
            }
            if (this.sysWebSocketServer.getServerKeys().contains(String.valueOf(request.getReceiverId()))) {
                this.sysWebSocketServer.getWebSocket(String.valueOf(request.getReceiverId())).send(JavaSerializableUtils.serializable(request));
            }
        } else {
            throw new Exception("request receiverId:" + request.getReceiverId() + " is not exist");
        }
        this.responseFutureMap.put(request.getRequestId(), new ResponseFuture(request));
        return this.responseFutureMap.get(request.getRequestId()).get();
    }

    @Override
    public List<Response> sendAll(Request request) throws Exception {
        ResponseAllFuture responseAllFuture = new ResponseAllFuture(request, new AtomicInteger(getActiveSize()));
        request.setSenderConfig(nodeConfig);
        for (SysWebSocketClient sysWebSocketClient : this.sysWebSocketClients.values()) {
            if (sysWebSocketClient.isOpen()){
                sysWebSocketClient.send(JavaSerializableUtils.serializable(request));
            }
        }
        this.sysWebSocketServer.sendAll(JavaSerializableUtils.serializable(request));
        this.responseAllFutureMap.put(request.getRequestId(), responseAllFuture);
        return this.responseAllFutureMap.get(request.getRequestId()).get();
    }

    @Override
    public int getActiveSize() {
        return getActiveKeys().size();
    }

    @Override
    public List<String> getActiveKeys() {
        List<String> serverKeys = new ArrayList<>();
        for (Long serverId : this.sysWebSocketClients.keySet()) {
            SysWebSocketClient sysWebSocketClient = this.sysWebSocketClients.get(serverId);
            if (sysWebSocketClient.isOpen()) {
                serverKeys.add(serverId.toString());
            }
        }
        if (!this.sysWebSocketServer.getServerKeys().isEmpty()) {
            serverKeys.addAll(this.sysWebSocketServer.getServerKeys());
        }
        return serverKeys;
    }
}
